package ds_industry_2025.ds.ds_03.T5

import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import java.util.Properties
import scala.collection.mutable.ArrayBuffer
/*
    1、根据子任务一的结果，对其进行SVD分解，对数据进行降维保留前5个奇异值信息，根据该用户已购买的商品分别与未购买的商品计算余弦相
    似度再进行累加求均值，将均值最大的5件商品id进行输出作为推荐使用。将输出结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】
    中对应的任务序号下。
结果格式如下：

------------------------推荐Top5结果如下------------------------
相似度top1(商品id：1，平均相似度：0.983456)
相似度top2(商品id：71，平均相似度：0.782672)
相似度top3(商品id：22，平均相似度：0.7635246)
相似度top4(商品id：351，平均相似度：0.7335748)
相似度top5(商品id：14，平均相似度：0.522356)
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparKSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    //  todo 读取特征工程生成的结果
    val data = spark.table("tzgc.skumat")

    //  todo 将除了用户id字段，把用户的购买和未购买数据拿到
    val skus:Seq[Double] = data.filter(col("user_id")===0.0).limit(1)
      .select(data.columns.slice(1, data.columns.length).map(col): _*)
      .map(r => r.toSeq.map(r => r.toString.toDouble))
      .collect()(0)

    println("0.0用户的购买商品数据情况")
    println(skus.mkString(","))

    //  todo 创建两个数组分别存储商品的索引
    var buy= new ArrayBuffer[Int]()
    val no_buy=new ArrayBuffer[Int]()

    //  todo 将商品的索引添加到数组
    skus.zipWithIndex.foreach{
      e =>{
        if(e._1.equals(0.0)){
          //  如果该数据未购买的话，就索引添加到no_buy
          no_buy += e._2
        }else{
          //  该商品是购买的，将该商品的索引添加到buy
          buy += e._2
        }
      }
    }


    //  todo 将读取到特征工程里面的表，除了id字段以外，将每一行数据转化为一个密集向量
    //  todo 为了方便后面将数据转化为rowMatrix(做svd分解需要)
    //  todo Vectors.dense()需要传一个array[Double]的参数，所以结尾需要转化为数组
    val array:Array[Vector] = data.rdd.map(
      r => {
        //  todo 这里返回的数据类型是Vectors是因为Vectors是工厂类，denseVectors和SparseVectors都是抽象类
        val vector: Vector = Vectors.dense(r.toSeq.slice(1, r.toSeq.length).map(r => r.toString.toDouble).toArray)
        vector
      }
    ).collect()

    //  todo 将数据转化为rowMatrix
    val matrix = new RowMatrix(spark.sparkContext.parallelize(array))
    // todo 对rowMatrix进行svd分解，保留前5个奇异值，并且计算U
    //  computeU = true：表示计算左奇异向量矩阵U。如果不需要U，可以设置为false以节省计算资源
    //  返回一个SingularValueDecomposition对象，包含三个部分：U（左奇异向量矩阵）、Σ（奇异值矩阵）、V（右奇异向量矩阵）
    val svd = matrix.computeSVD(5, computeU = true)
    //  拿到右奇异向量矩阵
    val v:Matrix=svd.V

    //  todo 将得到右奇异向量矩阵，添加索引，转化为数组
    val toArray:Array[(Vector,Int)] = v.rowIter.zipWithIndex.toArray

    //  todo 将带索引的奇异向量矩阵转化为dataframe
    val sku_info = spark.createDataFrame(
      toArray
    ).toDF("vec", "sku_mapping_id")

    //  todo 注册计算余弦相似度的函数
    spark.udf.register(
      "cos",
      (v1:DenseVector,v2:DenseVector) =>{
        1 - breeze.linalg.functions.cosineDistance(
          breeze.linalg.DenseVector(v1.values.toArray),
          breeze.linalg.DenseVector(v2.values.toArray)
        )
      }
    )

    //  使用交叉连接,计算平均余弦相似度
    val result = sku_info.crossJoin(sku_info)
      .toDF("vec", "sku_mapping_id", "vec2", "sku_mapping_id2")
      .withColumn("cos", expr("cos(vec,vec2)"))
      .filter(col("vec") !== col("vec2"))
      .filter(col("sku_mapping_id").isin(buy: _*) && col("sku_mapping_id2").isin(no_buy: _*))
      .groupBy("sku_mapping_id2")
      .agg(
        avg("cos").as("avg_cos")
      )
      .orderBy(desc("avg_cos"))
      .limit(5)


    println("-----------------推荐Top5结果如下-----------------")
    result.collect().zipWithIndex.foreach{
      case (r,index) =>{
        val id=r.getAs[Int]("sku_mapping_id2")
        val cos=r.getAs[Double]("avg_cos")
        val str=s"相似度top${index+1}(商品id:${id},平均相似度:${cos})"
        println(str)
      }
    }

















    spark.close()
  }

}
